package com.zero.rocketmq.api.order;

import com.zero.rocketmq.api.utils.RocketMQContents;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * 顺序消息，生产者
 */
public class OrderProducer {

    public static void main(String[] args) throws MQClientException {

        DefaultMQProducer producer = new DefaultMQProducer(RocketMQContents.PRODUCER_GROUP);
        producer.setNamesrvAddr(RocketMQContents.NAMESERVER_ADDR);
        producer.start();

        try {

            for (int i = 0; i < 10; i++) {
                int orderId = i;
                for (int j = 0; j<5; j++){
                    Message msg = new Message("OrderTopicTest",
                            "order_"+orderId,
                            "KEY" + orderId,
                            ("order_"+orderId+" step " + j).getBytes(RemotingHelper.DEFAULT_CHARSET));

                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            Integer id = (Integer) arg;
                            int index = id % mqs.size();
                            return mqs.get(index);
                        }
                    }, orderId);
                    System.out.printf("%s%n", sendResult);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            producer.shutdown();
        }
    }
}
